{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark Feedforward neural network example\") \\\n",
    "    .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----------------+-----------+--------------+---------+-------------------+--------------------+-------+----+---------+-------+-------+\n",
      "|fixed acidity|volatile acidity|citric acid|residual sugar|chlorides|free sulfur dioxide|total sulfur dioxide|density|  pH|sulphates|alcohol|quality|\n",
      "+-------------+----------------+-----------+--------------+---------+-------------------+--------------------+-------+----+---------+-------+-------+\n",
      "|          7.4|             0.7|        0.0|           1.9|    0.076|               11.0|                34.0| 0.9978|3.51|     0.56|    9.4|      5|\n",
      "|          7.8|            0.88|        0.0|           2.6|    0.098|               25.0|                67.0| 0.9968| 3.2|     0.68|    9.8|      5|\n",
      "|          7.8|            0.76|       0.04|           2.3|    0.092|               15.0|                54.0|  0.997|3.26|     0.65|    9.8|      5|\n",
      "|         11.2|            0.28|       0.56|           1.9|    0.075|               17.0|                60.0|  0.998|3.16|     0.58|    9.8|      6|\n",
      "|          7.4|             0.7|        0.0|           1.9|    0.076|               11.0|                34.0| 0.9978|3.51|     0.56|    9.4|      5|\n",
      "+-------------+----------------+-----------+--------------+---------+-------------------+--------------------+-------+----+---------+-------+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.classification import MultilayerPerceptronClassifier\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "\n",
    "# Load training data\n",
    "df = spark.read.format('com.databricks.spark.csv').\\\n",
    "                               options(header='true', \\\n",
    "                               inferschema='true').load(\"../data/WineData.csv\",header=True);\n",
    "df.show(5)    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- fixed acidity: double (nullable = true)\n",
      " |-- volatile acidity: double (nullable = true)\n",
      " |-- citric acid: double (nullable = true)\n",
      " |-- residual sugar: double (nullable = true)\n",
      " |-- chlorides: double (nullable = true)\n",
      " |-- free sulfur dioxide: double (nullable = true)\n",
      " |-- total sulfur dioxide: double (nullable = true)\n",
      " |-- density: double (nullable = true)\n",
      " |-- pH: double (nullable = true)\n",
      " |-- sulphates: double (nullable = true)\n",
      " |-- alcohol: double (nullable = true)\n",
      " |-- quality: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Convert to float format\n",
    "def string_to_float(x):\n",
    "    return float(x)\n",
    "\n",
    "# \n",
    "def condition(r):\n",
    "    if (0<= r <= 4):\n",
    "        label = \"low\" \n",
    "    elif(4< r <= 6):\n",
    "        label = \"medium\"\n",
    "    else: \n",
    "        label = \"high\" \n",
    "    return label"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType, DoubleType\n",
    "string_to_float_udf = udf(string_to_float, DoubleType())\n",
    "quality_udf = udf(lambda x: condition(x), StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#df= df.withColumn(\"quality\", string_to_float_udf(\"quality\")).withColumn(\"Cquality\", quality_udf(\"quality\"))\n",
    "df= df.withColumn(\"quality\", quality_udf(\"quality\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- fixed acidity: double (nullable = true)\n",
      " |-- volatile acidity: double (nullable = true)\n",
      " |-- citric acid: double (nullable = true)\n",
      " |-- residual sugar: double (nullable = true)\n",
      " |-- chlorides: double (nullable = true)\n",
      " |-- free sulfur dioxide: double (nullable = true)\n",
      " |-- total sulfur dioxide: double (nullable = true)\n",
      " |-- density: double (nullable = true)\n",
      " |-- pH: double (nullable = true)\n",
      " |-- sulphates: double (nullable = true)\n",
      " |-- alcohol: double (nullable = true)\n",
      " |-- quality: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "12"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----------------+-----------+--------------+---------+-------------------+--------------------+-------+----+---------+-------+-------+\n",
      "|fixed acidity|volatile acidity|citric acid|residual sugar|chlorides|free sulfur dioxide|total sulfur dioxide|density|  pH|sulphates|alcohol|quality|\n",
      "+-------------+----------------+-----------+--------------+---------+-------------------+--------------------+-------+----+---------+-------+-------+\n",
      "|          7.4|             0.7|        0.0|           1.9|    0.076|               11.0|                34.0| 0.9978|3.51|     0.56|    9.4| medium|\n",
      "|          7.8|            0.88|        0.0|           2.6|    0.098|               25.0|                67.0| 0.9968| 3.2|     0.68|    9.8| medium|\n",
      "|          7.8|            0.76|       0.04|           2.3|    0.092|               15.0|                54.0|  0.997|3.26|     0.65|    9.8| medium|\n",
      "|         11.2|            0.28|       0.56|           1.9|    0.075|               17.0|                60.0|  0.998|3.16|     0.58|    9.8| medium|\n",
      "|          7.4|             0.7|        0.0|           1.9|    0.076|               11.0|                34.0| 0.9978|3.51|     0.56|    9.4| medium|\n",
      "|          7.4|            0.66|        0.0|           1.8|    0.075|               13.0|                40.0| 0.9978|3.51|     0.56|    9.4| medium|\n",
      "|          7.9|             0.6|       0.06|           1.6|    0.069|               15.0|                59.0| 0.9964| 3.3|     0.46|    9.4| medium|\n",
      "|          7.3|            0.65|        0.0|           1.2|    0.065|               15.0|                21.0| 0.9946|3.39|     0.47|   10.0|   high|\n",
      "|          7.8|            0.58|       0.02|           2.0|    0.073|                9.0|                18.0| 0.9968|3.36|     0.57|    9.5|   high|\n",
      "|          7.5|             0.5|       0.36|           6.1|    0.071|               17.0|               102.0| 0.9978|3.35|      0.8|   10.5| medium|\n",
      "|          6.7|            0.58|       0.08|           1.8|    0.097|               15.0|                65.0| 0.9959|3.28|     0.54|    9.2| medium|\n",
      "|          7.5|             0.5|       0.36|           6.1|    0.071|               17.0|               102.0| 0.9978|3.35|      0.8|   10.5| medium|\n",
      "|          5.6|           0.615|        0.0|           1.6|    0.089|               16.0|                59.0| 0.9943|3.58|     0.52|    9.9| medium|\n",
      "|          7.8|            0.61|       0.29|           1.6|    0.114|                9.0|                29.0| 0.9974|3.26|     1.56|    9.1| medium|\n",
      "|          8.9|            0.62|       0.18|           3.8|    0.176|               52.0|               145.0| 0.9986|3.16|     0.88|    9.2| medium|\n",
      "|          8.9|            0.62|       0.19|           3.9|     0.17|               51.0|               148.0| 0.9986|3.17|     0.93|    9.2| medium|\n",
      "|          8.5|            0.28|       0.56|           1.8|    0.092|               35.0|               103.0| 0.9969| 3.3|     0.75|   10.5|   high|\n",
      "|          8.1|            0.56|       0.28|           1.7|    0.368|               16.0|                56.0| 0.9968|3.11|     1.28|    9.3| medium|\n",
      "|          7.4|            0.59|       0.08|           4.4|    0.086|                6.0|                29.0| 0.9974|3.38|      0.5|    9.0|    low|\n",
      "|          7.9|            0.32|       0.51|           1.8|    0.341|               17.0|                56.0| 0.9969|3.04|     1.08|    9.2| medium|\n",
      "+-------------+----------------+-----------+--------------+---------+-------------------+--------------------+-------+----+---------+-------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['label', 'features']"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# convert the data to dense vector\n",
    "def transData(data):\n",
    "    return data.rdd.map(lambda r: [r[-1], Vectors.dense(r[:-1])]).toDF(['label','features'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/Users/mingchen/anaconda2/lib/python2.7/site-packages/pytz/__init__.py:29: UserWarning: Module argparse was already imported from /Users/mingchen/anaconda2/lib/python2.7/argparse.pyc, but /Users/mingchen/anaconda2/lib/python2.7/site-packages/argparse-1.4.0-py2.7.egg is being added to sys.path\n",
      "  from pkg_resources import resource_stream\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+\n",
      "| label|            features|\n",
      "+------+--------------------+\n",
      "|medium|[7.4,0.7,0.0,1.9,...|\n",
      "|medium|[7.8,0.88,0.0,2.6...|\n",
      "|medium|[7.8,0.76,0.04,2....|\n",
      "|medium|[11.2,0.28,0.56,1...|\n",
      "|medium|[7.4,0.7,0.0,1.9,...|\n",
      "|medium|[7.4,0.66,0.0,1.8...|\n",
      "|medium|[7.9,0.6,0.06,1.6...|\n",
      "|  high|[7.3,0.65,0.0,1.2...|\n",
      "|  high|[7.8,0.58,0.02,2....|\n",
      "|medium|[7.5,0.5,0.36,6.1...|\n",
      "|medium|[6.7,0.58,0.08,1....|\n",
      "|medium|[7.5,0.5,0.36,6.1...|\n",
      "|medium|[5.6,0.615,0.0,1....|\n",
      "|medium|[7.8,0.61,0.29,1....|\n",
      "|medium|[8.9,0.62,0.18,3....|\n",
      "|medium|[8.9,0.62,0.19,3....|\n",
      "|  high|[8.5,0.28,0.56,1....|\n",
      "|medium|[8.1,0.56,0.28,1....|\n",
      "|   low|[7.4,0.59,0.08,4....|\n",
      "|medium|[7.9,0.32,0.51,1....|\n",
      "+------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import Row\n",
    "from pyspark.ml.linalg import Vectors\n",
    "\n",
    "data= transData(df)\n",
    "data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+------------+\n",
      "| label|            features|indexedLabel|\n",
      "+------+--------------------+------------+\n",
      "|medium|[7.4,0.7,0.0,1.9,...|         0.0|\n",
      "|medium|[7.8,0.88,0.0,2.6...|         0.0|\n",
      "|medium|[7.8,0.76,0.04,2....|         0.0|\n",
      "|medium|[11.2,0.28,0.56,1...|         0.0|\n",
      "|medium|[7.4,0.7,0.0,1.9,...|         0.0|\n",
      "|medium|[7.4,0.66,0.0,1.8...|         0.0|\n",
      "+------+--------------------+------------+\n",
      "only showing top 6 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import IndexToString, StringIndexer, VectorIndexer\n",
    "# Index labels, adding metadata to the label column.\n",
    "# Fit on whole dataset to include all labels in index.\n",
    "labelIndexer = StringIndexer(inputCol=\"label\", outputCol=\"indexedLabel\").fit(data)\n",
    "labelIndexer.transform(data).show(6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+--------------------+\n",
      "| label|            features|     indexedFeatures|\n",
      "+------+--------------------+--------------------+\n",
      "|medium|[7.4,0.7,0.0,1.9,...|[7.4,0.7,0.0,1.9,...|\n",
      "|medium|[7.8,0.88,0.0,2.6...|[7.8,0.88,0.0,2.6...|\n",
      "|medium|[7.8,0.76,0.04,2....|[7.8,0.76,0.04,2....|\n",
      "|medium|[11.2,0.28,0.56,1...|[11.2,0.28,0.56,1...|\n",
      "|medium|[7.4,0.7,0.0,1.9,...|[7.4,0.7,0.0,1.9,...|\n",
      "|medium|[7.4,0.66,0.0,1.8...|[7.4,0.66,0.0,1.8...|\n",
      "+------+--------------------+--------------------+\n",
      "only showing top 6 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Automatically identify categorical features, and index them.\n",
    "# Set maxCategories so features with > 4 distinct values are treated as continuous.\n",
    "featureIndexer =VectorIndexer(inputCol=\"features\", \\\n",
    "                              outputCol=\"indexedFeatures\", \\\n",
    "                              maxCategories=4).fit(data)\n",
    "\n",
    "featureIndexer.transform(data).show(6)   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- label: string (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Split the data into train and test\n",
    "(trainingData, testData) = data.randomSplit([0.6, 0.4])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+\n",
      "| label|            features|\n",
      "+------+--------------------+\n",
      "|medium|[7.4,0.7,0.0,1.9,...|\n",
      "|medium|[7.8,0.88,0.0,2.6...|\n",
      "|medium|[7.8,0.76,0.04,2....|\n",
      "|medium|[11.2,0.28,0.56,1...|\n",
      "|medium|[7.4,0.7,0.0,1.9,...|\n",
      "|medium|[7.4,0.66,0.0,1.8...|\n",
      "|medium|[7.9,0.6,0.06,1.6...|\n",
      "|  high|[7.3,0.65,0.0,1.2...|\n",
      "|  high|[7.8,0.58,0.02,2....|\n",
      "|medium|[7.5,0.5,0.36,6.1...|\n",
      "|medium|[6.7,0.58,0.08,1....|\n",
      "|medium|[7.5,0.5,0.36,6.1...|\n",
      "|medium|[5.6,0.615,0.0,1....|\n",
      "|medium|[7.8,0.61,0.29,1....|\n",
      "|medium|[8.9,0.62,0.18,3....|\n",
      "|medium|[8.9,0.62,0.19,3....|\n",
      "|  high|[8.5,0.28,0.56,1....|\n",
      "|medium|[8.1,0.56,0.28,1....|\n",
      "|   low|[7.4,0.59,0.08,4....|\n",
      "|medium|[7.9,0.32,0.51,1....|\n",
      "+------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# specify layers for the neural network:\n",
    "# input layer of size 11 (features), two intermediate of size 5 and 4\n",
    "# and output of size 7 (classes)\n",
    "layers = [11, 5, 4, 4, 3 , 7]\n",
    "\n",
    "# create the trainer and set its parameters\n",
    "FNN = MultilayerPerceptronClassifier(labelCol=\"indexedLabel\", featuresCol=\"indexedFeatures\",\\\n",
    "                                         maxIter=100, layers=layers, blockSize=128, seed=1234)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Convert indexed labels back to original labels.\n",
    "labelConverter = IndexToString(inputCol=\"prediction\", outputCol=\"predictedLabel\",\n",
    "                               labels=labelIndexer.labels)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Chain indexers and forest in a Pipeline\n",
    "from pyspark.ml import Pipeline\n",
    "pipeline = Pipeline(stages=[labelIndexer, featureIndexer, FNN, labelConverter])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# train the model\n",
    "# Train model.  This also runs the indexers.\n",
    "model = pipeline.fit(trainingData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Make predictions.\n",
    "predictions = model.transform(testData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+--------------+\n",
      "|            features|label|predictedLabel|\n",
      "+--------------------+-----+--------------+\n",
      "|[5.3,0.47,0.11,2....| high|        medium|\n",
      "|[6.4,0.31,0.09,1....| high|        medium|\n",
      "|[6.4,0.57,0.12,2....| high|        medium|\n",
      "|[6.6,0.56,0.14,2....| high|        medium|\n",
      "|[6.6,0.815,0.02,2...| high|        medium|\n",
      "+--------------------+-----+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Select example rows to display.\n",
    "predictions.select(\"features\",\"label\",\"predictedLabel\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Predictions accuracy = 0.839806, Test Error = 0.160194\n"
     ]
    }
   ],
   "source": [
    "# Select (prediction, true label) and compute test error\n",
    "evaluator = MulticlassClassificationEvaluator(\n",
    "    labelCol=\"indexedLabel\", predictionCol=\"prediction\", metricName=\"accuracy\")\n",
    "accuracy = evaluator.evaluate(predictions)\n",
    "print(\"Predictions accuracy = %g, Test Error = %g\" % (accuracy,(1.0 - accuracy)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [Root]",
   "language": "python",
   "name": "Python [Root]"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
